// 从mysql 同步数据,不是作为app的一部分,而作为一个单独的脚本执行
// node syn_data_from_mysql.js

"use strict";

var mysql = require('mysql');
var request = require('request');
var util = require('util');
var EventEmitter = require('events').EventEmitter;
var fs = require('fs');
var path = require('path');

var config_all = {
  dev: {
    url: 'http://localhost:1888/api/v1/article/logflow2',
    logFile: path.join(__dirname, 'mySqlsync.log'),
    cleanEsUrl: 'http://localhost:1888/api/v1/article/rebuild',
    mysql: {
      host: '192.168.61.10',
      port: 3306,
      user: 'csapp',
      database: 'mahoo',
      password: 'j1C9k3Mn'
    }
  },
  test: {
    url: 'http://192.168.61.10:1888/api/v1/article/logflow2',
    logFile: '/tmp/syn_mysql_to_es.log',
    cleanEsUrl: 'http://localhost:1888/api/v1/article/rebuild',
    mysql: {
      host: '192.168.61.10',
      port: 3306,
      user: 'csapp',
      database: 'mahoo',
      password: 'j1C9k3Mn'
    }
  },
  prod: {
    url: 'http://localhost:1888/api/v1/article/logflow2',
    logFile: '/tmp/syn_mysql_to_es.log',
    cleanEsUrl: 'http://localhost:1888/api/v1/article/rebuild',
    mysql: {
      host: '127.0.0.1',
      port: 3306,
      user: 'mahoo',
      database: 'mahoo',
      password: 'j3YcA9eEu0LqF'
    }
  }
};

var env = process.env.NODE_ENV || 'prod';
var __maxCount = -1; // -1表示全部
var __fromCount = 0;
var __pageSize = 1000; // 如果是一条一条插入，这个值为1


var url = config_all[env].url; // 一条一条插入，走完整流程
var cleanEsUrl = config_all[env].cleanEsUrl;
var logFile = config_all[env].logFile;
var mysqlConf = config_all[env].mysql;


// all event
// connected: connect to db succ
// count,size: get the count(*) size
// sqlResult,list : get the query result
// saveToEsDone, body : save success
function SyncMysqlData() {
}
util.inherits(SyncMysqlData, EventEmitter);
var theProto = SyncMysqlData.prototype;

/**
 *
 * @param maxCount
 * @param fromCount
 */
theProto.init = function (maxCount, fromCount, pageSize) {
  console.log('init');
  console.log(config_all[env]);
  this.maxCount = maxCount;
  this.fromCount = fromCount ? fromCount : 0;
  this.pageSize = pageSize ? pageSize : 500;
  this.stepInfo = {};
};

theProto.cleanEs = function () {
  var self = this;
  request(cleanEsUrl, function (err, resp, body) {
    if (err) throw err;
    console.log(body);
    self.emit('esCleaned');
  });
};

theProto.start = function () {
  console.log('start');
  this.emit('start');
  ////
  //var connection = mysql.createConnection(mysqlConf);
  //var self = this;
  //this.connection = connection;
  //connection.connect(function (err) {
  //  if (err) throw err;
  //  self.emit('connected')
  //});
};

theProto.connectMysql = function () {
  var connection = mysql.createConnection(mysqlConf);
  var self = this;
  this.connection = connection;
  connection.connect(function (err) {
    if (err) throw err;
    self.emit('connected')
  });
};

theProto.getCount = function () {
  //console.log('getCount');
  var self = this;
  var sql = 'select count(*) from mh_thread ';
  self.connection.query(sql, function (err, rows, fields) {
    var count = rows[0]["count(*)"];
    self.count = count;
    self.emit('count', count);
  });
};

theProto.nextPageIndex = function () {
  //console.log('nextPageIndex');
  if (!this.pageIndex) {
    this.pageIndex = 1;
  }
  var t = this.pageIndex;
  this.pageIndex++;
  return t;
};

theProto.logStep = function () {
  if (env === 'prod') {
    return;
  }
  var msg = '-----------------\n';
  for (var key in this.stepInfo) {
    msg += (key + ':' + this.stepInfo[key] + '\n');
  }
  this.stepInfo = {};
  //fs.appendFile(logFile, msg, function (err, res) {
  fs.writeFile(logFile, msg, function (err, res) {
    if (err) console.log(err);
  });
};

/**
 {
     "uid":"xxx",
     "articleId":"xxx",
     "nickName":"xxx",
     "articleContent":"xxx"
 }
 * @returns {*}
 */
theProto.getSelectSql = function () {
  //console.log('getSelectSql');
  var self = this;
  var offset = (this.nextPageIndex() - 1) * this.pageSize + this.fromCount;
  self.stepInfo.offset = offset;
  self.stepInfo.limit = this.pageSize;
  if (this.maxCount > 0 && offset - this.fromCount >= this.maxCount) {
    console.log('reach maxCount');
    return null;
  }
  if (offset > this.count) {
    console.log('reach all');
    return null;
  }
  var sql =
    //       uid,      vest,   tid,   cid,[ cid2, ctype,] timeline, lastpost,  views,   liketimes,      status,    replies,   userreplies
    "select t1.uid, t1.imgcount, t1.vest, t1.tid, t1.cid, t1.timeline, t1.lastpost, t1.views, t1.liketimes, t1.status, t1.replies, t1.userreplies, t1.pid, t1.ctype, " +
    " t2.post " + //articleContent
      // from
    " from mh_thread as t1, " +
    " mh_post_content as t2 " +
      // where
    " where t2.pid = t1.pid " +
      // limit offset
    " limit " + this.pageSize + ' offset ' + offset;

  return sql;
};

/**
 {
     "uid":"xxx",
     "articleId":"xxx",
     "nickName":"xxx",
     "articleContent":"xxx"
 }
 * @returns {*}
 */
theProto.queryForResult = function () {
  //console.log('queryForResult');
  var self = this;
  var sql = self.getSelectSql();
  if (!sql) {
    self.emit('done');
    return;
  }
  console.log(sql);
  self.connection.query(sql, function (err, rows, fields) {
    if (err) throw err;
    var list = [];
    self.stepInfo.getCount = rows.length;
    rows.forEach(function (item) {
      //if (item.post.length > 0) list.push(item.post);
      var obj = {
        // action: 'post',
        // uid, vest, tid, articleContent,cid,cid2,ctype, timeline, lastpost,views, liketimes,status,replies,userreplies
        uid: item.uid, //用户id
        vest: item.vest, // 马甲id 可能为空
        tid: item.tid, // 帖子id
        articleContent: item.post,
        cid: item.cid, //圈子id
        timeline: item.timeline,
        cid2: -1, // 分类id TODO
        ctype: item.ctype, // 分类id
        imgcount: item.imgcount, // 图片数量
        liketimes: item.liketimes,
        lastpost: item.lastpost,
        views: item.views,
        status: item.status,
        replies: item.replies,
        userreplies: item.userreplies,
      };
      //console.log(obj);
      list.push(obj);
    });
    self.emit('sqlResult', list);
  });
};


theProto.saveToEs = function (list) {
  //console.log('saveToEs');
  var self = this;
  request({
    method: 'POST',
    url: url,
    json: true,
    headers: {"content-type": "application/json"},
    body: {list: list}
  }, function (err, response, body) {
    if (err) {
      throw err;
    }
    //console.log(body);
    self.stepInfo.save = 'success';
    self.emit('saveToEsDone', body);
  });
};


theProto.stop = function () {
  console.log('stop');
  this.connection.end();
  this.emit('stop');
};

/********************/
/*      start       */
/********************/
var syn = new SyncMysqlData();
syn.init(__maxCount, __fromCount, __pageSize); // (maxCount, fromCount, pageSize)
// connected: connect to db succ
// count,size: get the count(*) size
// sqlResult,list : get the query result
// saveToEsDone, body : save success
// done : all done
// stop : connection over

syn.on('start', function () {
  this.cleanEs();
});

syn.on('esCleaned', function () {
  this.connectMysql();
});

syn.on('connected', function () {
  this.getCount();
});

syn.on('count', function (count) {
  console.log('count=' + count);
  this.queryForResult();
});
syn.on('sqlResult', function (list) {
  //console.log(list[0]);
  this.saveToEs(list);
});
syn.on('saveToEsDone', function () {
  var self = this;
  self.logStep();
  if (env == 'prod') {

    setTimeout(function () {
      self.queryForResult();
    }, 100); // 100ms per request
  } else {
    process.nextTick(function () {
      self.queryForResult();
    });
  }
});
syn.on('done', function () {
  this.stop();
});

syn.start();


